package ds_industry_2025.ds.ds_03.tzgc

import org.apache.spark.sql.functions._
import org.apache.spark.sql.{SparkSession, functions}
/*
    2、根据第1小题的结果，对其进行聚合，其中对sku_id进行one-hot转换，将其转换为以下格式矩阵：第一列为用户id，其余列名为商品id，
    按照用户id进行升序排序，展示矩阵第一行前5列数据，将结果截图粘贴至客户端桌面【Release\任务C提交结果.docx】中对应的任务序号下。
 */
object t2 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("t2")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    import spark.implicits._

    // todo 读取子任务一的结果
    val r1 = spark.table("tzgc.t2")

    // todo 对sku_id字段的值进行更换
    val source = r1.withColumn(
      "sku_id", functions.concat(lit("sku_id"), col("sku_id"))
    )

    //  todo 拿到所有的sku_id
    val skus:Array[String] = source.select("sku_id").distinct()
      .orderBy(
        split(col("sku_id"), "id")(1).cast("int")
      )
      .map(_(0).toString)
      .collect()

    //  todo 进行one-hot转换
    val result = source.groupBy("user_id")
      .pivot("sku_id", skus)
      .agg(lit(1.0))
      .na.fill(0.0)
      .orderBy("user_id")

    result.show
    //  todo 将结果保存到hive,给下面的推荐系统使用
    result.write.format("hive").mode("overwrite")
      .saveAsTable("tzgc.skumat")

    println("-------------第一行前5列结果展示为----------------")
  result.limit(1).select(result.columns.take(5).map(col): _*)
    .foreach(
      r => println(r.mkString(","))
    )

    spark.close()
  }

}
